package com.github.slprj.slake.net;

import java.nio.ByteBuffer;
import java.util.ArrayDeque;

/**
 * 写数据等待队列
 * -编码消数据体,并封装成buf
 * -避免buf创建的开销, 使用buffer pool 代替,用完后归还
 *
 * @author YangXiaoLong
 * 2022/9/17 20:02
 */
public class IoSubQueue {
    /**
     * 5k
     */
    private static final int MAX_BUFFER_SIZE = 512 * 1024;
    private final ByteBufferPool pool;
    private Runnable registerForWrite;
    private ByteBuffer writeBuffer;
    //todo dt_error
    private final ArrayDeque<WriteFrame> subQueue = new ArrayDeque<>();
    /**
     * 入队时,计算总的数据的大小
     */
    private int subQueueBytes;
    private boolean writing;

    public IoSubQueue(ByteBufferPool pool) {
        this.pool = pool;
    }

    public void setRegisterForWrite(Runnable registerForWrite) {
        this.registerForWrite = registerForWrite;
    }

    public void enqueue(WriteFrame frame) {
        ArrayDeque<WriteFrame> subQueue = this.subQueue;
        subQueue.add(frame);
        subQueueBytes += frame.estimateSize();
        if (subQueue.size() == 1 && !writing) {
            //触发写事件
            registerForWrite.run();
        }
    }

    public ByteBuffer getWriteBuffer(long nanoTime) {
        ByteBuffer writeBuffer = this.writeBuffer;
        if (writeBuffer != null) {
            //可能存在没有写完的情况,则进行写消息
            if (writeBuffer.remaining() > 0) {
                return writeBuffer;
            } else {
                pool.release(writeBuffer, nanoTime);
                this.writeBuffer = null;
            }
        }
        int subQueueBytes = this.subQueueBytes;
        //没有数据可以写
        if (subQueueBytes == 0) {
            return null;
        }
        ArrayDeque<WriteFrame> subQueue = this.subQueue;
        ByteBuffer buf = null;
        WriteFrame f;
        if (subQueueBytes <= MAX_BUFFER_SIZE) {
            buf = pool.borrow(subQueueBytes);
            while ((f = this.subQueue.pollFirst()) != null) {
                f.encode(buf);
            }
            this.subQueueBytes = 0;
        } else {
            while ((f = this.subQueue.pollFirst()) != null) {
                int size = f.estimateSize();
                if (buf == null) {
                    if (size > MAX_BUFFER_SIZE) {
                        buf = pool.borrow(size);
                        f.encode(buf);
                        subQueueBytes -= size;
                        break;
                    } else {
                        buf = pool.borrow(MAX_BUFFER_SIZE);
                    }
                }
                if (size > buf.remaining()) {
                    subQueue.addFirst(f);
                    break;
                } else {
                    f.encode(buf);
                    subQueueBytes -= size;
                }
            }
            this.subQueueBytes = subQueueBytes;
        }
        if (buf == null) {
            return null;
        }
        buf.flip();
        this.writeBuffer = buf;
        return buf;

    }

    public void setWriting(boolean writing) {
        this.writing = writing;
    }
}
